package ex.tableapi.window;

import ex.tableapi.ApiFrame;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.Tumble;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.rowInterval;

public class ExTumblingWindows extends ApiFrame {
    private String outName = "printOutTable";

    public static void main(String[] args) {
        ExTumblingWindows demo = new ExTumblingWindows();
        demo.getEnv();
        Table orders2 = demo.registerMysql2("myorder1", "orders1");
        Table result = demo.TumblingWindows(orders2);

        demo.registerConsole(demo.createPrintOutDDL(result.getResolvedSchema().toString()), demo.outName);
        result.executeInsert(demo.outName);
    }

    /**
     * 滚动窗口将行分配给固定长度的非重叠、连续窗口。例如，一个 5 分钟的滚动窗口以 5 分钟的间隔对行进行分组。可以在事件时间、处理时间或行计数上定义滚动窗口
     * see https://nightlies.apache.org/flink/flink-docs-release-1.17/zh/docs/dev/table/concepts/time_attributes/
     * @param orders
     * @return
     *  Table result = orders
     *                 .window(Tumble.over(lit(5).minutes()).on($("rowtime")).as("w")) // define window
     *                 .groupBy($("a"), $("w")) // group by key and window
     *                 // access window properties and aggregate
     *                 .select(
     *                         $("a"),
     *                         $("w").start().as("b1") ,
     *                         $("w").end().as("b2"),
     *                         $("w").rowtime().as("b3"),
     *                         $("b").sum().as("d")
     *                 );
     */
    private Table TumblingWindows(Table orders) {
        orders.printSchema();
        //TODO 不支持,在sql写？ https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/dev/table/concepts/time_attributes/
        Table result = orders.window(Tumble.over(rowInterval(2L)).on($("pt")).as("w"))
                .groupBy($("a"), $("w"))
                .select($("a"));
        //在rowtime 定义10分钟的滚动窗口并给窗口起一个别名w
//        Table result1 = orders.window(Tumble.over(lit(35).minutes()).on($("rowtime")).as("w")) //在rowtime 定义5分钟的滚动窗口并给窗口起一个别名w
//                .groupBy($("a"), $("w")) // group by key and window
//                .select(
//                        $("a"),
//                        $("w").start(),
//                        $("w").end(),
//                        $("w").rowtime(),
//                        $("b").count()
//                );

        return result;
    }


}
